package com.leilei.join.window;

import com.leilei.join.common.*;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/27 16:52
 * @desc fink 流相连之WindowJoin  (基于窗口的多流相连) 注意，测试发现无界流之间才能join
 * 滚动窗口未设置事件时间
 * 最终车辆数据:6> (1,保时捷卡宴,20210327,黄)
 * 最终车辆数据:6> (1,保时捷卡宴,20210327,黄)
 * 最终车辆数据:8> (3,阿斯顿马丁,20210327,宝石蓝)
 * 最终车辆数据:8> (3,阿斯顿马丁,20210327,宝石蓝)
 * 最终车辆数据:6> (1,保时捷卡宴,20210327,黄)
 * 最终车辆数据:8> (2,路虎揽胜,20210327,黑)
 * 最终车辆数据:8> (2,路虎揽胜,20210327,黑)
 */
public class FlinkWindowJoin {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 获取车辆数据
        DataStreamSource<Vehicle> vehicleDs = env.addSource(new VehicleSource());
        // 获取车辆明细数
        DataStreamSource<VehicleDetail> vehicleDetailDs = env.addSource(new VehicleDetailSource());
        // 车辆数据与 车辆明细数据关联，滚动五秒统计一次五秒内 关联情况
        DataStream<VehicleInfo> vehicleInfoDs = vehicleDs.join(vehicleDetailDs)
                .where(Vehicle::getId).equalTo(VehicleDetail::getVehicleId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply(new VehicleJoinFunction());
        //再将 vehicleInfo 与车辆类型数据关联
        DataStreamSource<Tuple2<Integer, String>> vehicleTypeDs = env.addSource(new VehicleTypeSource());
        DataStream<Tuple4<Integer, String, Integer, String>> finalSource = vehicleInfoDs.join(vehicleTypeDs)
                .where(VehicleInfo::getType).equalTo(t -> t.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(2)))
                .apply(new AggregationVehicleJoinFunction());

        finalSource.print("最终车辆数据");

        env.execute("window-join");

    }


}
